package com.tang.transform;

import com.tang.bean.WaterSensor;
import com.tang.functions.FilterFunctionImpl;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 过滤测试
 *
 * @author tang
 * @since 2023/6/7 10:07
 */
public class FilterDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("q2", 2L, 1),
                new WaterSensor("w3", 3L, 3),
                new WaterSensor("e4", 4L, 4)
        );

        // 1. 匿名内部类
        source.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor value) throws Exception {
                return "s1".equals(value.id);
            }
        }).print();
        
        // 2. lambda
        source.filter(waterSensor -> "s1".equals(waterSensor.getId())).print();
        
        // 3. 传入FilterFunction的实现类
        source.filter(new FilterFunctionImpl("s1")).print();

        env.execute();
    }

}
